package com.mjf.gmall.realtime.dim

import com.alibaba.fastjson.JSON
import com.mjf.gmall.realtime.bean.dim.BaseTradeMark
import com.mjf.gmall.realtime.util.{MyKafkaUtil, OffsetManager}
import org.apache.hadoop.conf.Configuration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.phoenix.spark._
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 实时维护品牌维度表
 *
 * 技术栈：
 * zookeeper/kafka/redis/hadoop/HBase/phoenix
 *
 * phoenix建表语句：
 * create table gmall0105_base_trademark(id varchar primary key, tm_name varchar);
 */
object TrademarkApp {
  def main(args: Array[String]): Unit = {

    // 读取Kafka中品牌信息的topic
    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("TrademarkApp")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topic = "ods_base_trademark"
    val groupId = "base_trademark_group"

    // 从redis读取offset
    val kafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    // 从Kafka消费数据
    var recordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (kafkaOffsetMap == null && kafkaOffsetMap.size > 0) {
      // 如果redis中已经有了offset信息
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, kafkaOffsetMap, groupId)
    } else {
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }

    //获得偏移结束点
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange] // 写在算子外面为启动时执行
    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = recordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 写在算子里面是周期性执行（driver端）
        rdd
    }


    // 写入到HBase中
    inputGetOffsetDstream.foreachRDD {
      rdd =>
        val baseTradeMarkRDD: RDD[BaseTradeMark] = rdd.map {
          record =>
            val jsonString: String = record.value()
            val baseTradeMark: BaseTradeMark = JSON.parseObject(jsonString, classOf[BaseTradeMark])
            baseTradeMark
        }

        baseTradeMarkRDD.saveToPhoenix(
          "GMALL0105_BASE_TRADEMARK",
          Seq("ID", "TM_NAME"),
          new Configuration,
          Some("hadoop102,hadoop103,hadoop104:2181")
        )

        // 数据处理结束，偏移量提交到redis
        OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    ssc.start()
    ssc.awaitTermination()

  }
}
